-- created by kumiler
-- on 2022/11/16 15:07
-- DESC
set hive.exec.dynamic.partition=true;            --  动态分区
set hive.exec.dynamic.partition.mode=nonstrict;
set hive.exec.max.dynamic.partitions=100;         --  每天生成 60 个分区
set hive.exec.max.dynamic.partitions.pernode=100; --  每天生成 60 个分区
with net as (
    select name,
           code,
           parent_network_code, -- 加盟商/中心：代理区；网点：加盟商
           parent_network_name,
           nvl(agent_code, financial_center_code) as agent_code,
           nvl(agent_name, financial_center_desc) as agent_name,
           manage_code,         -- bigint comment '大区ID',
           manage_name,         -- string comment '大区名称',
           network_type,
           if(network_type = 4, 1, 0)             as is_distribution_center
    from jms_dim.dim_network_whole_massage
    where instr(name, '测试') = 0
      and instr(code, 'test') = 0
),
     shipment_msg as (
         select shipment_no,
                start_code,
                end_code,
                max(shipment_name)       as shipment_name,
                min(actual_arrival_time) as actual_arrival_time,
                max(start_name)          as start_name,
                max(end_name)            as end_name
         from jms_dwd.dwd_tmsnew_shipment_union_base_dt
         where dt >= date_add('{{ execution_date | cst_ds }}', -5)
         group by shipment_no, start_code, end_code
     ),

     pre_network as (
         select t1.shipment_no,t1.sort_num,t1.stop_network_code,t1.stop_network_name
              ,concat_ws(',',collect_list(t2.network_code)) as pre_network_list
         from (
                  select shipment_no, sort_num, network_code as stop_network_code, network_name as stop_network_name
                  from jms_dwd.dwd_tmsnew_shipment_stop_union_base_dt
                  where dt >= date_add('{{ execution_date | cst_ds }}', -5) and shipment_no is not null
              ) t1 left join (
                  select shipment_no, sort_num, network_code
                  from jms_dwd.dwd_tmsnew_shipment_stop_union_base_dt
                  where dt >= date_add('{{ execution_date | cst_ds }}', -5) and shipment_no is not null
             ) t2
         on t1.shipment_no = t2.shipment_no and t1.sort_num < t2.sort_num
         group by t1.shipment_no,t1.sort_num,t1.stop_network_code,t1.stop_network_name
         having pre_network_list is not null
     ),

     -- 卸车
     fix_network as (
         select t1.shipment_no,t1.sort_num,t1.stop_network_code,t1.stop_network_name
              ,concat_ws(',',collect_list(t2.network_code)) as fix_network_list
         from (
                  select shipment_no, sort_num, network_code as stop_network_code, network_name as stop_network_name
                  from jms_dwd.dwd_tmsnew_shipment_stop_union_base_dt
                  where dt >= date_add('{{ execution_date | cst_ds }}', -5) and shipment_no is not null
              ) t1 left join (
                  select shipment_no, sort_num, network_code
                  from jms_dwd.dwd_tmsnew_shipment_stop_union_base_dt
                  where dt >= date_add('{{ execution_date | cst_ds }}', -5) and shipment_no is not null
             ) t2
         on t1.shipment_no = t2.shipment_no and t1.sort_num > t2.sort_num
         group by t1.shipment_no,t1.sort_num,t1.stop_network_code,t1.stop_network_name
         having fix_network_list is not null
     ),
     waybill_source as (
         select waybill_no as billcode,
                ordersource_code,
                ordersource_name,
                goods_type_code,
                goods_type_name
         from jms_dwd.dwd_yl_oms_oms_waybill_incre_dt
         where dt >= date_add('{{ execution_date | cst_ds }}', -5)
           and dt <= '{{ execution_date | cst_ds }}'
           and is_delete <> 2
           and (waybill_no like 'JT%' or waybill_no like '55%')
     ),
     -- 获取中心卸车扫描 T-5 12:00:00 -- T-1 12:00:00 的数据
     unloading as (
         select a.*
         from (
                  select waybill_no as billcode,       -- string comment '运单编号',
                         t0.shipment_no as transfercode,   -- string comment '转单号',
                         scan_type as scantype,       -- string comment '扫描类型',
                         transfer_center as transfercenter, -- string comment '中转中心',
                         pre_station as nextstation,    -- string comment '上一站或下一站',
                         scan_site_code as scansitecode,   -- string comment '扫描网点编号',
                         scan_time as scantime,       -- timestamp,扫描时间
                         min(scan_time) over(partition by t0.shipment_no,t0.scan_site_code) as early_scantime,
                         pre_station_code as nextstationcode,
                         goods_type as goodstype,
                         scan_user as scanuser,
                         scan_user_code as scanusercode,
                         pistol_id as pistolid,
                         row_number()
                                 over (partition by scan_site_code, waybill_no order by scan_time) as scantime_rank
                  from jms_dwd.dwd_tab_barscan_unloading_base_dt t0
                       -- 别问为啥是5天，产品提的
                  where dt >= date_add('{{ execution_date | cst_ds }}', -5)
                    and dt <= '{{ execution_date | cst_ds }}'
                    and scan_type = '卸车扫描'
                    and (waybill_no like 'JT%' or waybill_no like '55%')
                    and t0.shipment_no is not null 
              ) a
                  inner join
                  (select * from net where is_distribution_center = 1) n
                  on a.scansitecode = n.code
         where a.scantime_rank = 1
     ),
     loading as (
         select *
         from (
                  select waybill_no as billcode,        -- string,运单编号
                         t1.shipment_no as transfercode,    -- string,转单号
                         scan_type as scantype,        -- string,扫描类型
                         transf_ercenter as transfercenter,  -- string comment '中转中心',
                         next_station as nextstation,     -- string,上一站或下一站
                         scan_site_code as scansitecode,    -- string comment '扫描网点编号',
                         scan_time as scantime,        -- timestamp,扫描时间
                         next_station_code as nextstationcode, -- string,上下一站编号
                         goods_type as goodstype,
                         scan_user as scanuser,
                         scan_user_code as scanusercode,
                         pistol_id as pistolid,
                         row_number()
                                 over (partition by scan_site_code, waybill_no order by scan_time desc) as scantime_rank
                  from jms_dwd.dwd_tab_barscan_loading_base_dt t1
                       -- 取较卸车表早三天的数据
                  where dt >= date_add('{{ execution_date | cst_ds }}', -5) -- 向前取8天
                    and dt <= '{{ execution_date | cst_ds }}'
                    and scan_type = '装车扫描' 
                    and (waybill_no like 'JT%' or waybill_no like '55%')
                    and t1.shipment_no is not null
              ) a
         where scantime_rank = 1
     ),
     merge1 as (
         select t1.billcode        as billcode,
                waybill_source.ordersource_code,
                waybill_source.ordersource_name,
                waybill_source.goods_type_code,
                waybill_source.goods_type_name,
                t1.scantime        as unload_scantinme,
                t1.early_scantime  as unload_early_scantime,
                t1.scansitecode    as unload_scansitecode,
                t1.transfercode    as unload_transfercode,
                t1.nextstationcode as unload_nextstationcode,
                t1.scanusercode    as unload_scanusercode,
                t1.scanuser        as unload_scanuser,
                t1.pistolid        as unload_pistolid,
                t2.scansitecode    as load_scansitecode,
                t2.transfercode    as load_transfercode,
                t2.nextstationcode as load_nextstationcode,
                t2.scanusercode    as load_scanusercode,
                t2.scanuser        as load_scanuser,
                t2.pistolid        as load_pistolid
         from (
                  select billcode,
                         scantime,
                         scantype,
                         scansitecode,
                         transfercode,
                         nextstationcode,
                         scanusercode,
                         scanuser,
                         early_scantime,
                         pistolid
                  from unloading) t1
                  left join
              (select billcode,
                      scantime,
                      scantype,
                      scansitecode,
                      transfercode,
                      nextstationcode,
                      scanusercode,
                      scanuser,
                      pistolid
               from loading) t2
              on t1.billcode = t2.billcode and t1.scansitecode = t2.nextstationcode
                  left join waybill_source
                            on t1.billcode = waybill_source.billcode
         where t1.scantime > t2.scantime
           and t1.scansitecode is not null
           and t1.scantime >= concat(date_add('{{ execution_date | cst_ds }}',-1), ' 12:00:00')
           and t1.scantime < concat('{{ execution_date | cst_ds }}', ' 12:00:00')
     )
insert overwrite table jms_dm.dm_tms_waybill_wrong_shipment_dt partition (dt= '{{ execution_date | cst_ds }}')
select *
from (
         select merge1.billcode
              , merge1.ordersource_code
              , merge1.ordersource_name
              , merge1.goods_type_code
              , merge1.goods_type_name
              , net1.manage_code
              , net1.manage_name
              , net1.agent_code
              , net1.agent_name
              , load_scansitecode
              , net1.name                                                                                    as load_stationname
              , net1.network_type
              , load_transfercode                                                                            as load_shipment_no
              , ship1.shipment_name                                                                          as load_shipment_name
              , ship1.start_code                                                                             as loadship_start_code
              , ship1.end_code                                                                               as loadship_end_code
              , if(pre_network_list is not null and locate(unload_scansitecode, pre_network_list) > 0, 0, 1)   as load_wrong_shopment
              , load_scanusercode
              , load_scanuser
              , load_pistolid
              , net2.manage_code                                                                             as unload_manage_code
              , net2.manage_name                                                                             as unload_manage_name
              , net2.agent_code                                                                              as unload_agent_code
              , net2.agent_name                                                                              as unload_agent_name
              , unload_scansitecode                                                                          as unload_center_code
              , net2.name                                                                                    as unload_center_name
              , unload_transfercode                                                                          as unload_shipment_no
              , ship2.shipment_name                                                                          as unload_shipment_name
              , ship2.start_code                                                                             as unloadship_start_code
              , ship2.end_code                                                                               as unloadship_end_code
              , if(fix_network_list is not null and locate(load_scansitecode, fix_network_list) > 0, 0,1)    as unload_wrong_shopment
              , unload_scantinme
              , ship2.actual_arrival_time
              , unload_scanusercode
              , unload_scanuser
              , unload_pistolid
              , if(load_transfercode <> unload_transfercode, 1, 0)                                           as diff_shipment
              , if((pre_network_list is not null and locate(unload_scansitecode, pre_network_list) > 0)
                       or (fix_network_list is not null and  locate(load_scansitecode, fix_network_list) > 0), 0,1)   as wrong_shipment
              , '{{ execution_date | cst_ds }}'                                                                                 as date_time
              , unload_early_scantime
              , fix_network_list
              , pre_network_list
         from merge1
                  left join shipment_msg ship1 on merge1.load_transfercode = ship1.shipment_no
                  left join shipment_msg ship2 on merge1.unload_transfercode = ship2.shipment_no
                  left join net net1 on merge1.load_scansitecode = net1.code
                  left join net net2 on merge1.unload_scansitecode = net2.code
                  left join pre_network on merge1.load_transfercode = pre_network.shipment_no and merge1.load_scansitecode = pre_network.stop_network_code
                  left join fix_network on merge1.unload_transfercode = fix_network.shipment_no and merge1.unload_scansitecode = fix_network.stop_network_code
     ) t
where load_wrong_shopment = 1 or unload_wrong_shopment = 1
distribute by pmod(hash(rand()),20)
;